package demo.kafka.streams.cfg;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.kstream.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

@Configuration
public class MessageChannelCfg {

    private static final Logger logger = LoggerFactory.getLogger(MessageChannelCfg.class);

    @Bean("wordCountFunction")
    public Function<KStream<String, String>, KStream<String, Long>> wordCountFunction() {
        return textLines -> {
            // 处理逻辑与原始代码保持一致
            KStream<String, String> words = textLines.flatMapValues(
                    textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+"))
            );

            KGroupedStream<String, String> groupedWords = words.groupBy(
                    (key, word) -> word,
                    Grouped.with(Serdes.String(), Serdes.String())
            );

            KTable<String, Long> wordCounts = groupedWords.count(
                    Materialized.as("counts-store")
            );

            return wordCounts.toStream();
        };
    }

    @Bean
    public Supplier<String> textSupplier() {
        String text = "hello world,hello kafka, hello kafka streams, hello java, hello spring, hello spring cloud";
        String[] array = text.split(",");
        Random random = new Random();
        int idx = random.nextInt(array.length);
        String word = array[idx];
        logger.info("textSupplier 生产单词: {}", word);
        return () -> word;
    }

    @Bean("printMsgConsumer")
    public Consumer<KStream<String, Long>> printMsgConsumer() {
        return new Consumer<KStream<String, Long>>() {
            @Override
            public void accept(KStream<String, Long> kStream) {
                kStream.foreach((key, value) -> {
                    logger.info("=== 收到消息 ===, {} = {}", key, value);
                    System.err.println(String.format("=== 收到消息 ===, %s = %s", key, value));
                });
            }
        };
    }

//
//    @Bean
//    public BinderCustomizer binderCustomizer() {
//        return (binder, binderName) -> {
//            if (binder instanceof KafkaStreamsBinder kafkaStreamsBinder) {
//                // 获取原始配置并修改
//                KafkaStreamsBinderConfigurationProperties binderConfiguration =
//                        kafkaStreamsBinder.getBinderConfiguration();
//
//                binderConfiguration.setApplicationId("my-app-" + lane);
//
//                // 这里可以更精细地控制Binder的行为
//            }
//        };
//    }

}